package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import com.pig4cloud.pigx.contract.vo.AppContractSymbolIncreaseRankingVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 待优化，热点榜和涨跌榜拆分为两个不同的webSocket
 * <p>存在问题，例：涨跌榜类型从1切换到4后，依然在推送1的数据
 *
 * @author 13596
 */
@Slf4j
@Component
public class Ranking extends RankingAbstractTopic {

	private final MongoTemplate mongo;

	private final ContractUtils utils;

	protected String TOPICNAME = "ranking";


	public Ranking(MongoTemplate mongo, ContractUtils utils) {
		this.mongo = mongo;
		this.utils = utils;
		init();
	}

	private void init() {
		wsPush();
	}

	private void wsPush() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				// ----------该推送方案 仅限于推送信息与用户信息无关联的场景----------
				// ----------实现方式为 将所有session根据查询类型进行划分,大幅度减少查询消耗,可承受大面积用户同时访问----------
				// ----------在服务器允许的情况下 可以将每种数据类型都new一个线程去执行,可大幅度提升并发访问的推送效率----------
				// 数据分类容器
				Map<String, List<WebSocketSession>> map = new HashMap<>();
				// 用户原生请求容器
				Map<String, List<String>> paramsMap = new HashMap<>();
				// 统计所有订阅排行榜用户的订阅类型
				for (Map.Entry<WebSocketSession, List<AbstractTopicVO>> entry : subList.entrySet()) {
					if (!entry.getKey().isOpen()) {
						subList.remove(entry.getKey());
						log.info("ws 移除已关闭的session");
						continue;
					}

					// 遍历请求参数
					List<AbstractTopicVO> paramList = entry.getValue();
					paramList.forEach(param -> {
						// 请求参数解析
						List<String> dataList = param.getDataList();
						if (dataList.size() < 2) {
							return;
						}

						// 排行榜类型
						String key = dataList.get(1);
						// 如果已有该类型,则在该排行榜类型中添加发送用户
						if (map.containsKey(key)) {
							map.get(key).add(entry.getKey());
						} else {
							List<WebSocketSession> sessionList = new ArrayList<>();
							sessionList.add(entry.getKey());
							map.put(key, sessionList);
							// 保存用户原生请求
							paramsMap.put(key, dataList);
						}
					});
				}

				// 遍历所有需要发送的排行榜类型,将各类型数据分发给订阅的用户
				map.forEach((key, value) -> {
					// 根据类型获取排行榜数据
					Integer type = Integer.valueOf(key);
					JSONArray ranking = getRanking(type);
					TextMessage message = getTextMessage(paramsMap.get(key), ranking);
					// 将数据发送给所有订阅该排行榜的用户
					value.forEach(session -> {
						try {
							session.sendMessage(message);
						} catch (IOException e) {
							e.printStackTrace();
							log.info("push ranking err:" + e.getMessage());
						}
					});
				});
			} catch (Exception e) {
				e.printStackTrace();
				log.info("push ranking err:" + e.getMessage());
			}
		}, 0, 2000, TimeUnit.MILLISECONDS);
	}

	public void sendMessage(Map.Entry<WebSocketSession, AbstractTopicVO> entry) {
		// 遍历请求参数
		AbstractTopicVO paramList = entry.getValue();
		// 请求参数解析
		List<String> dataList = paramList.getDataList();
		if (dataList.size() < 2) {
			return;
		}
		// 排行榜类型
		String key = dataList.get(1);
		Integer type = Integer.valueOf(key);
		JSONArray ranking = getRanking(type);
		TextMessage message = getTextMessage(dataList, ranking);
		try {
			entry.getKey().sendMessage(message);
		} catch (IOException e) {
			log.info("push ranking err:" + e.getMessage());
		}
	}

	/**
	 * 根据排行榜类型获取排行榜数据
	 *
	 * @param type 类型
	 */
	public JSONArray getRanking(Integer type) {
		Query query = new Query();
		// 分页 排行榜默认查排行前十 ； 热点币全部
		query.skip(0);
		query.addCriteria(Criteria.where("status").is("1"));
		if (type == 1) {
			// 涨幅榜
			query.limit(10);
			query.with(Sort.by(Sort.Order.desc("riseAndFallRange")));
		} else if (type == 2) {
			// 跌幅榜
			query.limit(10);
			query.with(Sort.by(Sort.Order.asc("riseAndFallRange")));
		} else if (type == 3) {
			// 成交榜
			query.with(Sort.by(Sort.Order.desc("sumPrice")));
			query.limit(10);
		} else if (type == 4) {
			query.with(Sort.by(Sort.Order.asc("orderNo")));
			query.addCriteria(Criteria.where("isHot").is("1"));
		}
		List<AppContractSymbolIncreaseRankingVO> voList = mongo.find(query, AppContractSymbolIncreaseRankingVO.class);
		for (AppContractSymbolIncreaseRankingVO vo : voList) {
			utils.setScaleBySymbol(vo, vo.getSymbolName());
		}
		return JSONArray.parseArray(JSON.toJSONString(voList));
	}
}
